import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Properties;

/**********************************
 * @author wmx
 * @date 2019/11/28 10:00
 * @description:
 * @version 1.0
 ************************************/
public class ConsumerKafka {

    public static void  consumerData(){
        Properties props = new Properties();

        props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");

        props.put("group.id", "test");

        props.put("enable.auto.commit", "false");

        props.put("auto.commit.interval.ms", "1000");

        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String,	String> consumer	=	new

        KafkaConsumer<>(props);

        consumer.subscribe(Arrays.asList("test9"));

        while (true) {
            try {
                ConsumerRecords<String,	String> records	=
                        consumer.poll(1000);
                for (ConsumerRecord<String, String> record : records)
                    System.out.printf("offset	=  %d,	key  =  %s,	value = %s%n", record.offset(), record.key(), record.value());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        consumerData();
    }
};


